package app;

import com.alibaba.fastjson.JSON;
import domain.Access;
import domain.EventCatagoryProductCount;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.util.Collector;
import utils.TopNAggregateFunction;
import utils.TopNWindowFunction;

import java.util.ArrayList;
import java.util.List;


/**
 * Create by zhh on 2022/9/16.
 * 统计相同事件类型，商品分类，商品名称，窗口的事件数量前三
 */
public class TopNAppV1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Access> cleanData = env.readTextFile("data/access.json")
                .map(new MapFunction<String, Access>() {
            @Override
            public Access map(String s) throws Exception {
                //TODO... json==>Access
                try {
                    return JSON.parseObject(s, Access.class);
                } catch (Exception e) {
                    e.printStackTrace();
                    return null;
                }
            }
        })
                .filter(x -> x != null)
                .assignTimestampsAndWatermarks(//设置可延迟20秒
                        new BoundedOutOfOrdernessTimestampExtractor<Access>(Time.seconds(20)) {
                            @Override
                            public long extractTimestamp(Access element) {
                                return element.time;
                            }
                        }
                )
                .filter(new FilterFunction<Access>() {
                    @Override
                    //过滤启动的用户
                    public boolean filter(Access access) throws Exception {
                        return !"startup".equals(access.event);//过滤不是startup的
                    }
                });
        //事件类型  商品类别  商品名称
        SingleOutputStreamOperator<EventCatagoryProductCount> aggStream = cleanData.keyBy(new KeySelector<Access, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> getKey(Access value) throws Exception {
                return Tuple3.of(value.event, value.product.category, value.product.name);
            }
        })
                .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))//滑动窗口
                //windowFunction  开始做增量计数的计算，后面排序取TopN时需要用全量数据，则在此使用aggregate的一种方法
                .aggregate(new TopNAggregateFunction(), new TopNWindowFunction());

        /**
         * 将 {event='cart', category='食品饮料', product='五子', count=182, start='2026-01-30 03:40:00', end='2026-01-30 03:45:00'}
         * 按照event,category,start,end分组
         */
        KeyedStream<EventCatagoryProductCount, Tuple4<String, String, Long, Long>> keyedStream = aggStream
                .keyBy(new KeySelector<EventCatagoryProductCount, Tuple4<String, String, Long, Long>>() {

            @Override
            public Tuple4<String, String, Long, Long> getKey(EventCatagoryProductCount value) throws Exception {
                return Tuple4.of(value.event, value.category, value.start, value.end);
            }
        });
        SingleOutputStreamOperator<List<EventCatagoryProductCount>> sortedStream = keyedStream.process(new KeyedProcessFunction<Tuple4<String, String, Long, Long>, EventCatagoryProductCount, List<EventCatagoryProductCount>>() {
            //创建ListState
            private transient ListState<EventCatagoryProductCount> listState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //初始化ListState
                listState = getRuntimeContext().getListState(new ListStateDescriptor<EventCatagoryProductCount>("cnt-state", EventCatagoryProductCount.class));
            }

            @Override
            public void processElement(EventCatagoryProductCount value, Context ctx, Collector<List<EventCatagoryProductCount>> out) throws Exception {
                listState.add(value);

                //定时器，规定窗口结束后多一秒触发接下来的执行
                ctx.timerService().registerEventTimeTimer(value.end + 1);
            }

            //满足了时间条件，触发onTimer,完成TopN的操作
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<List<EventCatagoryProductCount>> out) throws Exception {
                ArrayList<EventCatagoryProductCount> list = Lists.newArrayList(listState.get());
                list.sort((x, y) -> Long.compare(y.count, x.count));//降序
                ArrayList<EventCatagoryProductCount> sort = new ArrayList<>();

                //取前三，如果list的大小大于3就取3,小于3就取list.size
                for (int i = 0; i < Math.min(3, list.size()); i++) {
                    EventCatagoryProductCount bean = list.get(i);
                    sort.add(bean);
                }
                out.collect(sort);
            }
        });
        /**
         * [{event='browse', category='车', product='奔驰', count=178, start='1769715420000', end='1769715720000'},
         * {event='browse', category='车', product='法拉利', count=172, start='1769715420000', end='1769715720000'},
         * {event='browse', category='车', product='奥迪', count=171, start='1769715420000', end='1769715720000'}]
         */
        sortedStream.print().setParallelism(1);

        env.execute("TopNApp");

    }
}